import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteAtomicSequence;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.SqlQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.zk.TcpDiscoveryZookeeperIpFinder;
import org.apache.ignite.transactions.Transaction;

import java.util.ArrayList;
import java.util.List;

/**
 * Baihejia
 * Created by wangguodong on 12/04/2018.
 */
public class IgniteHelper {

    static IgniteConfiguration cfg;

    static Ignite igniteClient;

    static IgniteCache cache;

    static IgniteAtomicSequence seq;

    static {
        System.setProperty("IGNITE_REST_START_ON_CLIENT", "true");
        System.setProperty("IGNITE_QUIET", "false");
        System.setProperty("PROP_TX_ALLOWED", "true");

        TcpDiscoverySpi spi = new TcpDiscoverySpi();
        TcpDiscoveryZookeeperIpFinder ipFinder = new TcpDiscoveryZookeeperIpFinder();
        ipFinder.setZkConnectionString("10.141.142.227:2181");
        spi.setIpFinder(ipFinder);

        cfg = new IgniteConfiguration();
        cfg.setDiscoverySpi(spi);
        cfg.setPeerClassLoadingEnabled(true);
        Ignition.setClientMode(true);
        igniteClient = Ignition.start(cfg);

        cache = Ignition.ignite().getOrCreateCache(new CacheConfiguration<>("Area").setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
//        cache = igniteClient.getOrCreateCache(new CacheConfiguration<>("Area").setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
        seq = igniteClient.atomicSequence("Area", 0, false);
    }

    public static <T extends Entity> T save(T entity) throws Exception {
        try {
            Class cls = entity.getClass();

            Object id = entity.id;

            if (id == null) {
                entity.setKey(seq.incrementAndGet());
            }

            cache.put(entity.id, entity);

            return entity;

//        System.out.println(JsonHelper.toString(igniteClient.getOrCreateCache("Area").query(new SqlFieldsQuery("select * from Area.Area where id=?").setArgs(1))));

//        City p = new City(1l, "北京", "北京", (byte)1);


//        cache.query(new SqlFieldsQuery("update City.City set name='北京' where id=?").setArgs(1));
//
//        System.out.println(JsonHelper.toString(cache.query(new SqlFieldsQuery("select * from City.City c, Area.Area a where c.id=a.cityId"))));

//        IgniteAtomicSequence seq = igniteClient.atomicSequence("Area", 0, false);
//
//        for (int i = 0; i < 10; i++) {
//            System.out.println(seq.incrementAndGet());
//        }
//
//        System.out.println(JsonHelper.toMap(igniteClient.getOrCreateCache("Community").query(new SqlFieldsQuery("select count(1) from Community.Community c, Area.Area a where c.areaId=a.id"))));
//        System.out.println("finished");
        } finally {
//            igniteClient.close();
        }
    }

    public static <T> T findById(Class<T> cls, Object id) throws Exception {
        return (T) cache.get(id);
    }

    public static <T> List<T> find(Class<T> cls, String sql, Object... args) throws Exception {
        List<T> list = new ArrayList<>();
        for (T t : (Iterable<T>) cache.query(new SqlQuery(cls, sql).setArgs(args))) {
            list.add(t);
        }
        return list;
    }

    public static List find(String sql, Object... args) throws Exception {
        List list = new ArrayList<>();
        for (Object t : cache.query(new SqlFieldsQuery(sql).setArgs(args))) {
            list.add(t);
        }
        return list;
    }

    public static void close() {
        igniteClient.close();
    }

    public static void withTransaction(Execute execute) throws Exception {
        try (Transaction tx = Ignition.ignite().transactions().txStart()) {
            execute.call();
            tx.commit();
        }
    }

    public abstract class Execute {
        public abstract void call() throws Exception;
    }
}
